package com.shujia.flink.core

import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable

object Demo7ProcessFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    /**
     * process: 是flink底层的一个算子，传入一条可以返回多条，再process里面可以操作flink的数据，时间和状态
     * 用法
     * 1、再普通的DataStream后面使用，需要一个ProcessFunction
     */
    val wordsDS: DataStream[String] = linesDS
      .process(new ProcessFunction[String, String] {
        /**
         * processElement：么一条数据执行一次，进来一条可以输出多条
         *
         * @param value :一行数据
         * @param ctx   ： 上下文对象
         * @param out   ： 用于将数据发送到下游
         */
        override def processElement(value: String,
                                    ctx: ProcessFunction[String, String]#Context,
                                    out: Collector[String]): Unit = {

          val split: Array[String] = value.split(",")
          //循环将数据发送到下游
          for (word <- split) {
            out.collect(word)
          }
        }
      })

    //安装单词分组
    val keyByDS: KeyedStream[String, String] = wordsDS
      .keyBy(word => word)

    /**
     * 2、再eyBy之后使用：需要一个KeyedProcessFunction
     */
    val countDS: DataStream[(String, Int)] = keyByDS
      .process(new KeyedProcessFunction[String, String, (String, Int)] {
        //通过一个task中的数据会共享同一个count
        //var count = 0

        //使用map集合保存单词的数量，为每一个单词保存一个数量，就不会冲突了
        val hashMap = new mutable.HashMap[String, Int]()

        /**
         * processElement: 每一条数据执行一次
         *
         * @param value ： 遗憾数据
         * @param ctx   ：上下文对象
         * @param out   ： 用于将数据发送到下游
         */
        override def processElement(value: String,
                                    ctx: KeyedProcessFunction[String, String, (String, Int)]#Context,
                                    out: Collector[(String, Int)]): Unit = {
          //统计单词的数量
          //count += 1

          //从hashmap中获取单词的数量
          //getOrElse: 如果key存在返回value,如果key不存在返回默认值
          var count: Int = hashMap.getOrElse(value, 0)

          //统计单词的数量
          count += 1

          //将新的单词的数量保存到hashMap中
          hashMap.put(value, count)

          //发送到处的数据到下游
          out.collect((value, count))
        }
      })

    //countDS.print()

    /**
     * 3、再window之后使用,需要一个ProcessWindowFunction
     */

    //划分窗口，每隔5秒一个窗口
    val windowDS: WindowedStream[String, String, TimeWindow] = keyByDS
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

    val windowCountDS: DataStream[(String, Int, Long, Long)] = windowDS
      .process(new ProcessWindowFunction[String, (String, Int, Long, Long), String, TimeWindow] {

        /**
         * process: 每一个窗口执行一次，会增和窗口的数据都传到process方法中
         *
         * @param key      ：分组的key
         * @param context  :上下文对象, 获取到窗口的开始和结束时间
         * @param elements ： 窗口内所有的数据
         * @param out      ：用于将数据发送到下游
         */
        override def process(key: String,
                             context: Context,
                             elements: Iterable[String],
                             out: Collector[(String, Int, Long, Long)]): Unit = {
          //计算窗口内单词的数量
          val count: Int = elements.size

          //从上下文中获取窗口的开始和结束时间
          val start: Long = context.window.getStart
          val end: Long = context.window.getEnd

          //将数据发送到下游
          out.collect((key, count, start, end))
        }
      })

    windowCountDS.print()

    env.execute()
  }
}
